package com.github.slprj.slake.net;

import com.github.slprj.slake.common.BitUtil;
import com.github.slprj.slake.log.SlLog;
import com.github.slprj.slake.log.SlLogs;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 待写数据暂存队列组件
 * -提供待写数据的暂存队列,所有的待写数据等待分配
 * -提供
 *
 * @author YangXiaoLong
 * 2022/8/26 17:42
 */
public class IoQueue {

    private static final SlLog log = SlLogs.getLogger(IoQueue.class);
    /**
     * ConcurrentLinkedQueue是一种非阻塞的线程安全队列,底层采用CAS保证线程的安全
     */
    private final ConcurrentLinkedQueue<WriteData> writeQueue = new ConcurrentLinkedQueue<>();
    private final ArrayList<SlChannel> channels;
    private final boolean server;
    private int invokeIndex;

    public IoQueue(ArrayList<SlChannel> channels) {
        this.channels = channels;
        this.server = channels == null;
    }

    public void write(WriteData obj) {
        writeQueue.add(obj);
    }

    public boolean dispatchWriteQueue(HashMap<Long, WriteData> pendingRequests) {
        ConcurrentLinkedQueue<WriteData> writeQueue = this.writeQueue;
        WriteData wd;
        boolean result = false;
        while ((wd = writeQueue.poll()) != null) {
            result |= enqueue(pendingRequests, wd);
        }
        return result;
    }

    private boolean enqueue(HashMap<Long, WriteData> pendingRequests, WriteData wd) {
        WriteFrame frame = wd.getData();
        SlChannel slc = wd.getSlc();
        if (slc == null) {
            Peer peer = wd.getPeer();
            if (peer == null) {
                if (!server && frame.getFrameType() == FrameType.TYPE_REQ) {
                    slc = selectChanel();
                    if (slc == null) {
                        wd.getFuture().completeExceptionally(new NetException("no available channel"));
                        return false;
                    }
                } else {
                    log.error("no peer set");
                    if (frame.getFrameType() == FrameType.TYPE_REQ) {
                        wd.getFuture().completeExceptionally(new NetException("no peer set"));
                    }
                    return false;
                }
            } else {
                slc = peer.getSlChannel();
                if (slc == null) {
                    if (frame.getFrameType() == FrameType.TYPE_REQ) {
                        wd.getFuture().completeExceptionally(new NetException("not connected"));
                    }
                    return false;
                }
            }
        }

        if (slc.isClosed()) {
            if (frame.getFrameType() == FrameType.TYPE_REQ) {
                wd.getFuture().completeExceptionally(new NetException("channel closed"));
            }
            return false;
        }
        if (frame.getFrameType() == FrameType.TYPE_REQ) {
            int seq = slc.getAndIncSeq();
            frame.setSeq(seq);
            long key = BitUtil.toLong(slc.getChannelIndexInWorker(), seq);
            WriteData old = pendingRequests.put(key, wd);
            if (old != null) {
                String errMsg = "dup seq: old=" + old.getData() + ", new=" + frame;
                log.error(errMsg);
                wd.getFuture().completeExceptionally(new NetException(errMsg));
                pendingRequests.put(key, old);
                return false;
            }
        }
        slc.getSubQueue().enqueue(frame);
        return true;
    }

    private SlChannel selectChanel() {
        //todo dongTing命名不规范,有待优化
        ArrayList<SlChannel> channels = this.channels;
        int size = channels.size();
        if (size == 0) {
            return null;
        }
        int idx = invokeIndex;
        if (idx < size) {
            invokeIndex = idx + 1;
            return channels.get(idx);
        }
        invokeIndex = 0;
        return channels.get(invokeIndex);
    }
}
